package kettle.example;

import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.core.row.RowMeta;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.*;

/**
 * @author lunfy
 * @description:
 * @date 2024-11-29 20:47
 */
public class KettleCustomStepProcessor extends BaseStep implements StepInterface {

    public KettleCustomStepProcessor(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta, Trans trans) {
        super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
    }

    private RowMetaInterface inputRowMeta = null;
    private RowMetaInterface outputRowMeta = null;

    @Override
    public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
        // 获取输入行
        Object[] inputRow = getRow();
        if (inputRow == null) {
            // 没有更多行了
            setOutputDone();
            return false;
        }

        // 判断是否第一行
        if (first) {
            first = false;

            // 获取输入行元数据
            this.inputRowMeta = getInputRowMeta();
            // 创建新的输出行元数据
            this.outputRowMeta = new RowMeta();

            if (inputRowMeta == null) {
                logBasic("inputRowMeta为null");
                return false;
            }

            // 添加除了"RN"之外的所有字段到输出行元数据
            for (int i = 0; i < inputRowMeta.size(); i++) {
                String columnName = inputRowMeta.getValueMeta(i).getName();
                if (!"RN".equalsIgnoreCase(columnName)) {
                    logBasic("列名：" + columnName);
                    outputRowMeta.addValueMeta(inputRowMeta.getValueMeta(i));
                } else {
                    logBasic("要排除的列名：" + columnName);
                }
            }
        }

        // 创建新的行数组, 长度为输出行元数据的字段数
        Object[] outputRow = RowDataUtil.createResizedCopy(inputRow, outputRowMeta.size());

        // 遍历输出行元数据，将输入行中对应字段的数据复制到输出行中
        for (ValueMetaInterface metaInterface : outputRowMeta.getValueMetaList()) {
            String columnName = metaInterface.getName();
            int inputRowColumnIndex = inputRowMeta.indexOfValue(columnName);
            int outputRowColumnIndex = outputRowMeta.indexOfValue(columnName);
            outputRow[outputRowColumnIndex] = inputRow[inputRowColumnIndex];
        }

        // 将处理过的行放入输出流
        putRow(outputRowMeta, outputRow);

        return true;
    }

}
